package com.tlb.springboot.starter.rocketmq.impl;

import com.aliyun.openservices.shade.com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageExt;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.tlb.springboot.starter.rocketmq.annotation.RocketMQListen;
import com.tlb.springboot.starter.trace.Constants;
import com.tlb.springboot.starter.trace.context.TraceContext;

import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * @author yuaneg 2018/8/30
 */
@Slf4j
public abstract class AbstractMessageListener {

    /**
     * spring 的bean对象
     */
    private Object bean;
    /**
     * 要被执行的method
     */
    private Method method;

    protected RocketMQListen rocketMQListen;

    protected int maxReconsumeTimes;

    private ObjectMapper objectMapper = new ObjectMapper();
    
    {
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    public AbstractMessageListener(Object bean, Method method, RocketMQListen rocketMQListen) {
        this.rocketMQListen = rocketMQListen;
        this.bean = bean;
        this.method = method;
        int maxParameterCounts = 2;
        if (method.getParameterCount() > maxParameterCounts || method.getParameterCount() == 0) {
            throw new RuntimeException("rocketMQ 监听参数个数最多为2个");
        }
        if (method.getParameterCount() == maxParameterCounts) {
            if (!method.getParameterTypes()[1].getSimpleName().equals(MessageExt.class.getSimpleName())) {
                throw new RuntimeException("监听方法的第二个参数必须为MessageExt");
            }
        }
        if (rocketMQListen.maxReconsumeTimes() == -1) {
            this.maxReconsumeTimes = 16;
        } else {
            this.maxReconsumeTimes = rocketMQListen.maxReconsumeTimes();
        }
    }


    protected void invokeMethod(MessageExt messageExt) throws Exception {
        try {
            TraceContext.setTraceId(messageExt.getUserProperty(Constants.TRACE_ID));
            long now = System.currentTimeMillis();
            Object value;
            if (method.getParameterTypes()[0].getSimpleName().equals(String.class.getSimpleName())) {
                value = new String(messageExt.getBody());
            } else {
                value = objectMapper.readValue(messageExt.getBody(), method.getParameterTypes()[0]);
            }
            if (method.getParameterCount() == 1) {
                method.invoke(bean, value);
            } else {
                method.invoke(bean, value, messageExt);
            }
            long costTime = System.currentTimeMillis() - now;
            Map<String, Object> jsonObject = new HashMap<>();
            jsonObject.put("rocketMQ", "消费成功");
            jsonObject.put("consume-time(ms)", costTime);
            jsonObject.put("message-content", new String(messageExt.getBody()));
            log.info(JSONObject.toJSONString(jsonObject));
        } catch (Exception e) {
            Map<String, Object> json = new LinkedHashMap<>();
            json.put("rocketMQ", "消费异常");
            json.put("topic", messageExt.getTopic());
            json.put("tags", messageExt.getTags());
            json.put("keys", messageExt.getKeys());
            json.put("body", new String(messageExt.getBody()));
            json.put("reConsumeTimes", messageExt.getReconsumeTimes());
            json.put("queueId", messageExt.getQueueId());
            json.put("msgId", messageExt.getMsgId());
            if (messageExt.getReconsumeTimes() >= this.maxReconsumeTimes || rocketMQListen.delayLevel() == -1) {
                log.error(JSONObject.toJSONString(json), e);
            } else {
                log.warn(JSONObject.toJSONString(json), e);
            }
            throw e;
        } finally {
            TraceContext.removeTraceId();
        }
    }
}
